package com.cxdr.awsips.packing.processor;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.cxdr.awsips.packing.bean.PackLog;
import com.cxdr.awsips.packing.config.SystemConfig;
import com.cxdr.awsips.packing.db.DBHelper;
import com.cxdr.awsips.packing.logger.LogFactory;
import com.cxdr.awsips.packing.pool.BaseQueue;

public class StorePackLogProccessor extends BaseProcessor {
	public static final BaseQueue<PackLog> LogQueue = new BaseQueue<PackLog>();
	private static Logger logger = LogFactory.makeNewLogger(StorePackLogProccessor.class);
	private static final int BATCH_SIZE = Integer.parseInt(SystemConfig.get("num.batchsize"));
	private StoreThread[] ths = new StoreThread[Integer.parseInt(SystemConfig.get("num.store-threads"))];
	private ExecutorService executor = Executors.newFixedThreadPool(ths.length);
	private static final String insertSql = "INSERT INTO t_pack_log(id,partition_month,create_time,source_file,target_file,file_type,create_day,create_hour,station_id)"
											+ " VALUES     (seq_t_pack_log.nextval,to_char(sysdate,'yyyy-mm'),sysdate,?,?,?,to_char(sysdate,'dd'),to_char(sysdate,'hh24'),?)";
	public StorePackLogProccessor(){
		for(int i =0;i<ths.length;i++){
			ths[i] = new StoreThread();
		}
	}
	private class StoreThread extends BaseProcessor {
		private Connection connection;

		StoreThread() {
			try {
				connection = DBHelper.getConnection();
				connection.setAutoCommit(false);
			} catch (SQLException e) {
				this.finish();
				logger.error("GetConnection Error", e);
				logger.error("StoreThread will not real running ....");
				finish();
				e.printStackTrace();
			}

		}

		@Override
		public void run() {
			logger.info("Start StoreReportProccessor : " + this.getId());
			PreparedStatement insertPs = null;
			// PreparedStatement selectPs = null;
			PreparedStatement updatePs = null;
			ResultSet rs = null;
			try {
				insertPs = connection.prepareStatement(insertSql);
				// selectPs = connection.prepareStatement(selectSql);
			} catch (SQLException e) {
				e.printStackTrace();
			}
			while (!isFinish()) {
				int count = 0;
				while (!LogQueue.isEmpty() && count < BATCH_SIZE) {
					try {
						PackLog log = LogQueue.take();
						if (log == null) {
							continue;
						}
						insertPs.setString(1,log.getSource_file());
						insertPs.setString(2, log.getTarget_file());
						insertPs.setString(3, log.getFile_type());
						insertPs.setString(4, log.getStation_ID());
						insertPs.addBatch();
					} catch (SQLException e) {
						e.printStackTrace();
					}
					count++;
				}
				if (count == 0) {
					continue;
				}
				try {
					logger.info("提交插入");
					insertPs.executeBatch();
					connection.commit();
					insertPs.clearBatch();
				} catch (SQLException e) {
					logger.error("插入到报错误", e);
					e.printStackTrace();
					try {
						connection.rollback();
					} catch (SQLException e1) {
						e1.printStackTrace();
					}
				}

				if (LogQueue.isEmpty()) {
					// finish();
					// StoreReportProccessor.this.finish();
					// logger.info("结束入库线程");
				}
			}
		}

	}

	@Override
	public void run() {
		logger.info("开始打包日志入库线程");
		for(StoreThread thread:ths){
			executor.submit(thread);
		}
	}

}
